package com.lg.monitor

import com.lg.bean.BusInfo
import org.apache.commons.dbcp.BasicDataSource
import org.apache.spark.sql.ForeachWriter

import java.sql.SQLException

object MysqlWriter {

  var driver = "com.mysql.jdbc.driver" //驱动名称
  var url = "jdbc:mysql://192.168.8.105:3306/lg_test" //你的mysql数据库表的路径
  var user = "root" //用户名
  var password = "12345678" //密码

  def getConnect(): BasicDataSource = {
    //设置必须的参数
    val ds = new BasicDataSource
    ds.setDriverClassName(driver)
    ds.setUrl(url)
    ds.setUsername(user)
    ds.setPassword(password)
    //设置连接池的管理策略参数
    ds.setInitialSize(2)
    ds.setMaxActive(100)
    System.out.println("数据库连接池初始化:driver:" + driver + ",url:" + url + ", user:" + user + ", password:" + password)
    ds
  }
}

class MysqlWriter extends ForeachWriter[BusInfo]{
  var ds: BasicDataSource = null
  override def open(partitionId: Long, epochId: Long): Boolean = {
    ds = MysqlWriter.getConnect()
    false
  }

  override def process(busInfo: BusInfo): Unit = {
    val statement = ds.getConnection().prepareStatement(
      s""""
         |insert into bus_info (
         |    deploy_num, sim_num, transport_num, plate_num, lg_lat, speed, direction
         |) values (${busInfo.deployNum}, ${busInfo.simNum}, ${busInfo.transportNum}, ${busInfo.plateNum}, ${busInfo.lglat}, ${busInfo.speed}, ${busInfo.direction})
         |""".stripMargin
    )
    statement.executeUpdate()
  }

  override def close(errorOrNull: Throwable): Unit = {
    try if (ds != null) ds.close()
    catch {
      case e: SQLException =>
        e.printStackTrace()
    }
  }
}
